package window
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.WindowFunction
import org.apache.flink.streaming.api.windowing.windows.GlobalWindow
import org.apache.flink.util.Collector

import java.text.SimpleDateFormat
import scala.util.Random

class WindowDemoCount {

  /*def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment
    val data: DataStream[String] = env.socketTextStream("linux121", 7777)
    val sdf = new SimpleDateFormat("HH:mm:ss.sss")
    val maped: DataStream[(String, String, String)] = data.map((value: String) => {
      val time = System.currentTimeMillis()
      val dateTime = sdf.format(time)
      val random = Random.nextInt(5)
      (value, dateTime, String.valueOf(random))
    })

    val keyed = maped.keyBy((value: (String, String, String)) => value._1)
    val countWindow: WindowedStream[(String, String, String), String, GlobalWindow] = keyed.countWindow(3, 1)
    val applyed = countWindow.apply(new WindowFunction[(String, String, String), String, String, GlobalWindow] {
      override def apply(key: String, window: GlobalWindow, input: Iterable[(String, String, String)], out: Collector[String]): Unit = {

        val sb = new StringBuilder()
        input.foreach(it => {
          sb.append(it._1 + ".." + it._2 + ".." + it._3)
        })
        out.collect(sb.toString())
      }
    })

  }*/

}
